#include <avro.h>
#include <stdio.h>
#include <stdlib.h>
#include <curl/curl.h>

#ifdef DEFLATE_CODEC
#define QUICKSTOP_CODEC  "deflate"
#else
#define QUICKSTOP_CODEC  "null"
#endif

avro_schema_t packet_schema;
avro_schema_t msg_body_schema;
#define BODY_MAX_LEN 1024
//#define OK_VALUE_COUNT 2
#define OK_VALUE_COUNT 20000

int count = 0;

/* A simple schema for our tutorial */

/*http·¢ËÍÏûÏ¢ÌåµÄschema*/
const char DOCS_SCHEMA[] = "{\
	\"type\":\"record\",\
	\"name\":\"docs\",\
	\"fields\":[\
		{\
			\"name\":\"doc_schema_name\",\
			\"type\":\"string\"\
		},\
		{\
			\"name\":\"doc_desc\",\
			\"type\":[{\
				\"type\":\"map\",\
				\"values\":\"string\"\
			},\"null\"]\
		},\
		{\
			\"name\":\"user_desc\",\
			\"type\":[{\
				\"type\":\"map\",\
				\"values\":\"string\"\
			},\"null\"]\
		},\
		{\
			\"name\":\"doc_set\",\
			\"type\":{\
				\"type\":\"array\",\
				\"items\":\"bytes\"\
			}\
		}\
	]\
}";
#if 1
const char DOC_SCHEMA[] = "{\
	\"type\": \"record\",\
	\"name\": \"doc\",\
	\"fields\": [\
		{\
			\"name\": \"boolean_s\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": \"boolean\"\
			}, \"null\"]\
		},\
		{\
			\"name\": \"boolean_a\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": {\
					\"type\": \"array\",\
					\"items\": \"boolean\"\
				}\
			}, \"null\"]\
		},\
		{\
			\"name\": \"int_s\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": \"int\"\
			}, \"null\"]\
		},\
		{\
			\"name\": \"int_a\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": {\
					\"type\": \"array\",\
					\"items\": \"int\"\
				}\
			}, \"null\"]\
		},\
		{\
			\"name\": \"long_s\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": \"long\"\
			}, \"null\"]\
		},\
		{\
			\"name\": \"long_a\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": {\
					\"type\": \"array\",\
					\"items\": \"long\"\
				}\
			}, \"null\"]\
		},\
		{\
			\"name\": \"float_s\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": \"float\"\
			}, \"null\"]\
		},\
		{\
			\"name\": \"float_a\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": {\
					\"type\": \"array\",\
					\"items\": \"float\"\
				}\
			}, \"null\"]\
		},\
		{\
			\"name\": \"double_s\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": \"double\"\
			}, \"null\"]\
		},\
		{\
			\"name\": \"double_a\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": {\
					\"type\": \"array\",\
					\"items\": \"double\"\
				}\
			}, \"null\"]\
		},\
		{\
			\"name\": \"bytes_s\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": \"bytes\"\
			}, \"null\"]\
		},\
		{\
			\"name\": \"bytes_a\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": {\
					\"type\": \"array\",\
					\"items\": \"bytes\"\
				}\
			},\"null\"]\
		},\
		{\
			\"name\": \"string_s\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": \"string\"\
			}, \"null\"]\
		},\
		{\
			\"name\": \"string_a\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": {\
					\"type\": \"array\",\
					\"items\": \"string\"\
				}\
			}, \"null\"]\
		},\
		{\
			\"name\": \"doc_s\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": \"doc\"\
			}, \"null\"]\
		},\
		{\
			\"name\": \"doc_a\",\
			\"type\": [{\
				\"type\": \"map\",\
				\"values\": {\
					\"type\": \"array\",\
					\"items\": \"doc\"\
				}\
			}, \"null\"]\
		}\
	]\
}";

#endif

void avro_init_doc_schema(avro_schema_t* doc_schema) {
    if (avro_schema_from_json_literal(DOC_SCHEMA, doc_schema)) {
        fprintf(stderr, "Unable to parse packet schema:%s\n", avro_strerror());
        exit(EXIT_FAILURE);
    }
}

void avro_init_docs_schema(avro_schema_t* docs_schema) {
    if (avro_schema_from_json_literal(DOCS_SCHEMA, docs_schema)) {
        fprintf(stderr, "Unable to parse msgbody schema:%s\n", avro_strerror());
        exit(EXIT_FAILURE);
    }
}

int avro_add_doc(avro_schema_t doc_schema, avro_datum_t data_array) {
    char buf[BODY_MAX_LEN] = {0};
    avro_writer_t writer;

    int64_t datalen = 0;

    avro_schema_t boolean_map_schema = avro_schema_map(avro_schema_boolean());
    avro_datum_t boolean_map_datum = avro_map(boolean_map_schema);
    avro_datum_t boolean_map_value = avro_boolean(0);
    avro_map_set(boolean_map_datum, "field1", boolean_map_value);

    avro_schema_t boolean_a_map_schema = avro_schema_map(avro_schema_array(avro_schema_boolean()));
    avro_datum_t boolean_a_map_datum = avro_map(boolean_a_map_schema);

    avro_schema_t int_map_schema = avro_schema_map(avro_schema_int());
    avro_datum_t int_map_datum = avro_map(int_map_schema);
    avro_datum_t int_map_value = avro_int32(234);
    avro_map_set(int_map_datum, "field2", int_map_value);

    avro_schema_t int_a_map_schema = avro_schema_map(avro_schema_array(avro_schema_int()));
    avro_datum_t int_a_map_datum = avro_map(int_a_map_schema);

    avro_schema_t long_map_schema = avro_schema_map(avro_schema_long());
    avro_datum_t long_map_datum = avro_map(long_map_schema);
    avro_datum_t long_map_value = avro_int64(6897798888);
    avro_map_set(long_map_datum, "field3", long_map_value);

    avro_schema_t long_a_map_schema = avro_schema_map(avro_schema_array(avro_schema_long()));
    avro_datum_t long_a_map_datum = avro_map(long_a_map_schema);

    avro_schema_t float_map_schema = avro_schema_map(avro_schema_float());
    avro_datum_t float_map_datum = avro_map(float_map_schema);
    avro_datum_t float_map_value = avro_float(123.56);
    avro_map_set(float_map_datum, "field4", float_map_value);

    avro_schema_t float_a_map_schema = avro_schema_map(avro_schema_array(avro_schema_float()));
    avro_datum_t float_a_map_datum = avro_map(float_a_map_schema);

    avro_schema_t double_map_schema = avro_schema_map(avro_schema_double());
    avro_datum_t double_map_datum = avro_map(double_map_schema);
    avro_datum_t double_map_value = avro_double(123.56);
    avro_map_set(double_map_datum, "field5", double_map_value);

    avro_schema_t double_a_map_schema = avro_schema_map(avro_schema_array(avro_schema_double()));
    avro_datum_t double_a_map_datum = avro_map(double_a_map_schema);

    avro_schema_t bytes_map_schema = avro_schema_map(avro_schema_bytes());
    avro_datum_t bytes_map_datum = avro_map(bytes_map_schema);

    avro_schema_t bytes_a_map_schema = avro_schema_map(avro_schema_array(avro_schema_bytes()));
    avro_datum_t bytes_a_map_datum = avro_map(bytes_a_map_schema);

    avro_schema_t string_map_schema = avro_schema_map(avro_schema_string());
    avro_datum_t string_map_datum = avro_map(string_map_schema);
    avro_datum_t string_map_value_1 = avro_string("thisaaa");
    avro_map_set(string_map_datum, "field6", string_map_value_1);
    avro_datum_t string_map_value_2 = avro_string("thisbbb");
    avro_map_set(string_map_datum, "field7", string_map_value_2);

    avro_schema_t string_a_map_schema = avro_schema_map(avro_schema_array(avro_schema_string()));
    avro_datum_t string_a_map_datum = avro_map(string_a_map_schema);

    avro_schema_t doc_map_schema = avro_schema_map(packet_schema);
    avro_datum_t doc_map_datum = avro_map(doc_map_schema);

    avro_schema_t doc_a_map_schema = avro_schema_map(avro_schema_array(packet_schema));
    avro_datum_t doc_a_map_datum = avro_map(doc_a_map_schema);

    /*
     * just check it,delete some
     */
#if 1
    if (avro_array_size(string_map_datum) != 2) {
        fprintf(stderr, "Unexpected map size\n");
        exit(EXIT_FAILURE);
    }
#endif
    /*
     * 组装整个结构，为最后的序列化做准备
     */

#if 1
    avro_schema_t boolean_union_schema = avro_schema_union();
    avro_schema_t int_union_schema = avro_schema_union();
    avro_schema_t long_union_schema = avro_schema_union();
    avro_schema_t float_union_schema = avro_schema_union();
    avro_schema_t double_union_schema = avro_schema_union();
    avro_schema_t bytes_union_schema = avro_schema_union();
    avro_schema_t string_union_schema = avro_schema_union();
    avro_schema_t doc_union_schema = avro_schema_union();

    avro_schema_t boolean_a_union_schema = avro_schema_union();
    avro_schema_t int_a_union_schema = avro_schema_union();
    avro_schema_t long_a_union_schema = avro_schema_union();
    avro_schema_t float_a_union_schema = avro_schema_union();
    avro_schema_t double_a_union_schema = avro_schema_union();
    avro_schema_t bytes_a_union_schema = avro_schema_union();
    avro_schema_t string_a_union_schema = avro_schema_union();
    avro_schema_t doc_a_union_schema = avro_schema_union();

    avro_schema_union_append(boolean_union_schema, boolean_map_schema);
    avro_schema_union_append(boolean_union_schema, avro_schema_null());

    avro_schema_union_append(boolean_a_union_schema, boolean_a_map_schema);
    avro_schema_union_append(boolean_a_union_schema, avro_schema_null());

    avro_schema_union_append(int_union_schema, int_map_schema);
    avro_schema_union_append(int_union_schema, avro_schema_null());

    avro_schema_union_append(int_a_union_schema, int_a_map_schema);
    avro_schema_union_append(int_a_union_schema, avro_schema_null());

    avro_schema_union_append(long_union_schema, long_map_schema);
    avro_schema_union_append(long_union_schema, avro_schema_null());

    avro_schema_union_append(long_a_union_schema, long_a_map_schema);
    avro_schema_union_append(long_a_union_schema, avro_schema_null());

    avro_schema_union_append(float_union_schema, float_map_schema);
    avro_schema_union_append(float_union_schema, avro_schema_null());

    avro_schema_union_append(float_a_union_schema, float_a_map_schema);
    avro_schema_union_append(float_a_union_schema, avro_schema_null());

    avro_schema_union_append(double_union_schema, double_map_schema);
    avro_schema_union_append(double_union_schema, avro_schema_null());

    avro_schema_union_append(double_a_union_schema, double_a_map_schema);
    avro_schema_union_append(double_a_union_schema, avro_schema_null());

    avro_schema_union_append(bytes_union_schema, bytes_map_schema);
    avro_schema_union_append(bytes_union_schema, avro_schema_null());

    avro_schema_union_append(bytes_a_union_schema, bytes_a_map_schema);
    avro_schema_union_append(bytes_a_union_schema, avro_schema_null());

    avro_schema_union_append(string_union_schema, string_map_schema);
    avro_schema_union_append(string_union_schema, avro_schema_null());

    avro_schema_union_append(string_a_union_schema, string_a_map_schema);
    avro_schema_union_append(string_a_union_schema, avro_schema_null());

    avro_schema_union_append(doc_union_schema, doc_map_schema);
    avro_schema_union_append(doc_union_schema, avro_schema_null());

    avro_schema_union_append(doc_a_union_schema, doc_a_map_schema);
    avro_schema_union_append(doc_a_union_schema, avro_schema_null());

    avro_datum_t boolean_union_datum;
    avro_datum_t int_union_datum;
    avro_datum_t long_union_datum;
    avro_datum_t float_union_datum;
    avro_datum_t double_union_datum;
    avro_datum_t bytes_union_datum;
    avro_datum_t string_union_datum;
    avro_datum_t doc_union_datum;

    avro_datum_t boolean_a_union_datum;
    avro_datum_t int_a_union_datum;
    avro_datum_t long_a_union_datum;
    avro_datum_t float_a_union_datum;
    avro_datum_t double_a_union_datum;
    avro_datum_t bytes_a_union_datum;
    avro_datum_t string_a_union_datum;
    avro_datum_t doc_a_union_datum;


    if (count >= OK_VALUE_COUNT) {
        boolean_union_datum = avro_union(boolean_union_schema, 0, avro_null());
        int_union_datum = avro_union(int_union_schema, 0, avro_null());
        float_union_datum = avro_union(float_union_schema, 0, avro_null());
        double_union_datum = avro_union(double_union_schema, 0, avro_null());
        long_union_datum = avro_union(long_union_schema, 0, avro_null());
        bytes_union_datum = avro_union(bytes_union_schema, 0, avro_null());
        string_union_datum = avro_union(string_union_schema, 0, avro_null());
        doc_union_datum = avro_union(doc_union_schema, 0, avro_null());

        boolean_a_union_datum = avro_union(boolean_a_union_schema, 0, avro_null());
        int_a_union_datum = avro_union(int_a_union_schema, 0, avro_null());
        float_a_union_datum = avro_union(float_a_union_schema, 0, avro_null());
        double_a_union_datum = avro_union(double_a_union_schema, 0, avro_null());
        long_a_union_datum = avro_union(long_a_union_schema, 0, avro_null());
        bytes_a_union_datum = avro_union(bytes_a_union_schema, 0, avro_null());
        string_a_union_datum = avro_union(string_a_union_schema, 0, avro_null());
        doc_a_union_datum = avro_union(doc_a_union_schema, 0, avro_null());
    } else {
        boolean_union_datum = avro_union(boolean_union_schema, 0, boolean_map_datum);
        int_union_datum = avro_union(int_union_schema, 0, int_map_datum);
        float_union_datum = avro_union(float_union_schema, 0, float_map_datum);
        double_union_datum = avro_union(double_union_schema, 0, double_map_datum);
        long_union_datum = avro_union(long_union_schema, 0, long_map_datum);
        bytes_union_datum = avro_union(bytes_union_schema, 0, bytes_map_datum);
        string_union_datum = avro_union(string_union_schema, 0, string_map_datum);
        doc_union_datum = avro_union(doc_union_schema, 0, doc_map_datum);

        boolean_a_union_datum = avro_union(boolean_a_union_schema, 0, boolean_a_map_datum);
        int_a_union_datum = avro_union(int_a_union_schema, 0, int_a_map_datum);
        float_a_union_datum = avro_union(float_a_union_schema, 0, float_a_map_datum);
        double_a_union_datum = avro_union(double_a_union_schema, 0, double_a_map_datum);
        long_a_union_datum = avro_union(long_a_union_schema, 0, long_a_map_datum);
        bytes_a_union_datum = avro_union(bytes_a_union_schema, 0, bytes_a_map_datum);
        string_a_union_datum = avro_union(string_a_union_schema, 0, string_a_map_datum);
        doc_a_union_datum = avro_union(doc_a_union_schema, 0, doc_a_map_datum);
    }
#endif
    count++;

    avro_datum_t single_doc = avro_datum_from_schema(doc_schema);

    if (avro_record_set(single_doc, "boolean_s", boolean_union_datum)) {
        fprintf(stderr, "boolean map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "boolean_a", boolean_a_union_datum)) {
        fprintf(stderr, "boolean_a map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "int_s", int_union_datum)) {
        fprintf(stderr, "int map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "int_a", int_a_union_datum)) {
        fprintf(stderr, "int_a map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "long_s", long_union_datum)) {
        fprintf(stderr, "long map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "long_a", long_a_union_datum)) {
        fprintf(stderr, "long_a map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "float_s", float_union_datum)) {
        fprintf(stderr, "float map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "float_a", float_a_union_datum)) {
        fprintf(stderr, "float_a map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "double_s", double_union_datum)) {
        fprintf(stderr, "double map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "double_a", double_a_union_datum)) {
        fprintf(stderr, "double_a map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "bytes_s", bytes_union_datum)) {
        fprintf(stderr, "bytes map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "bytes_a", bytes_a_union_datum)) {
        fprintf(stderr, "bytes_a map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "string_s", string_union_datum)) {
        fprintf(stderr, "string map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "string_a", string_a_union_datum)) {
        fprintf(stderr, "string_a map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "doc_s", doc_union_datum)) {
        fprintf(stderr, "doc map error message: %s\n", avro_strerror());
        return 1;
    }

    if (avro_record_set(single_doc, "doc_a", doc_a_union_datum)) {
        fprintf(stderr, "doc_a map error message: %s\n", avro_strerror());
        return 1;
    }

    writer = avro_writer_memory(buf, BODY_MAX_LEN);
    if (writer == NULL) {
        printf("avro_writer_memory error\n");
        exit(0);
    }
#if 1
    if ((avro_write_data(writer, doc_schema, single_doc)) == EINVAL) {
        printf("avro_write_data error\n");
        perror(avro_strerror());
        exit(0);
    }
#endif
    avro_writer_flush(writer);
    datalen = avro_writer_tell(writer);

    fprintf(stderr, "data length is %d\n", datalen);

    avro_datum_t bytes_datum = avro_bytes(buf, datalen);
    avro_array_append_datum(data_array, bytes_datum);
    avro_datum_decref(bytes_datum);

    avro_schema_decref(string_map_schema);
    avro_datum_decref(string_map_datum);
    avro_datum_decref(string_map_value_1);
    avro_datum_decref(string_map_value_2);

    avro_schema_decref(long_map_schema);
    avro_datum_decref(long_map_datum);
    avro_datum_decref(long_map_value);
#if 1
    avro_schema_decref(string_union_schema);
    avro_datum_decref(string_union_datum);
    avro_schema_decref(long_union_schema);
    avro_datum_decref(long_union_datum);
#endif


    avro_datum_decref(single_doc);

    avro_writer_free(writer);

    fprintf(stderr, "data length is %d\n", datalen);
    return 0;
}

int avro_add_msgbody(avro_schema_t msgbody_schema,
        const char * name,
        avro_schema_t sschema,
        avro_datum_t doc_desc,
        avro_schema_t sschema_1,
        avro_datum_t user_desc,
        avro_datum_t data_array,
        avro_datum_t msg_body) {
    avro_datum_t name_datum = avro_string(name);

#if 0
    avro_schema_t string_union_schema = avro_schema_union();
    avro_schema_union_append(string_union_schema, sschema);
    avro_schema_union_append(string_union_schema, avro_schema_null());
    avro_datum_t strdesc_union_datum;
    strdesc_union_datum = avro_union(string_union_schema, 0, doc_desc);
    avro_schema_decref(string_union_schema);

    avro_schema_union_append(string_union_schema, sschema_1);
    avro_schema_union_append(string_union_schema, avro_schema_null());
    avro_datum_t struser_union_datum;
    struser_union_datum = avro_union(string_union_schema, 0, user_desc);
#else
    avro_schema_t string_map_schema = avro_schema_map(avro_schema_string());
    avro_schema_t string_union_schema = avro_schema_union();
    avro_schema_union_append(string_union_schema, string_map_schema);
    avro_schema_union_append(string_union_schema, avro_schema_null());
    avro_datum_t strdesc_union_datum;
    strdesc_union_datum = avro_union(string_union_schema, 0, doc_desc);
    avro_datum_t struser_union_datum;
    struser_union_datum = avro_union(string_union_schema, 0, user_desc);
//    struser_union_datum = avro_union(string_union_schema, 1, avro_null());
#endif
    if (avro_record_set(msg_body, "doc_schema_name", name_datum)
            || avro_record_set(msg_body, "doc_desc", strdesc_union_datum)
            || avro_record_set(msg_body, "user_desc", struser_union_datum)
            || avro_record_set(msg_body, "doc_set", data_array)) {
        fprintf(stderr, " error message: %s\n", avro_strerror());
        return 1;
    }

    avro_datum_decref(name_datum);
    avro_schema_decref(string_union_schema);
    avro_datum_decref(strdesc_union_datum);
    avro_datum_decref(struser_union_datum);
    return 0;
}

int main(void) {
    char buf[102400];


    int i = 0;
    while (i < 1) {
        i++;
        memset(buf, 0, strlen(buf));
        avro_init_doc_schema(&packet_schema);
        avro_init_docs_schema(&msg_body_schema);
        avro_schema_t array_schema = avro_schema_array(avro_schema_bytes());
        avro_datum_t data_array = avro_datum_from_schema(array_schema);
        int i = 0;
        for (i = 0; i < 1; i++) {
            avro_add_doc(packet_schema, data_array);
        }

        //doc desc ,  map 的value 为 string 类型
        avro_schema_t string_map_schema = avro_schema_map(avro_schema_string());
        avro_datum_t string_map = avro_map(string_map_schema);
        avro_datum_t map_value = avro_string("boolean_s");
        avro_map_set(string_map, "field1", map_value);
        avro_datum_decref(map_value);

        map_value = avro_string("int_s");
        avro_map_set(string_map, "field2", map_value);
        avro_datum_decref(map_value);

        map_value = avro_string("long_s");
        avro_map_set(string_map, "field3", map_value);
        avro_datum_decref(map_value);
        
        map_value = avro_string("long_s");
        avro_map_set(string_map, "field3", map_value);
        avro_datum_decref(map_value);
        
        map_value = avro_string("float_s");
        avro_map_set(string_map, "field4", map_value);
        avro_datum_decref(map_value);
        
        map_value = avro_string("double_s");
        avro_map_set(string_map, "field5", map_value);
        avro_datum_decref(map_value);
        
        map_value = avro_string("string_s");
        avro_map_set(string_map, "field6", map_value);
        avro_datum_decref(map_value);
        
        map_value = avro_string("string_s");
        avro_map_set(string_map, "field7", map_value);
        avro_datum_decref(map_value);

        avro_schema_t strusr_map_schema = avro_schema_map(avro_schema_string());
        avro_datum_t string_map_1 = avro_map(strusr_map_schema);
        map_value = avro_string("lmy");
        avro_map_set(string_map_1, "author", map_value);

        avro_datum_t body_record = avro_record(msg_body_schema);
        //avro_datum_t  body_record  = avro_datum_from_schema(packet_schema) ;
        avro_add_msgbody(msg_body_schema, "db1.testwlwlwl", string_map_schema, string_map, strusr_map_schema, string_map_1, data_array, body_record);
        //Serialization
        avro_writer_t writer = avro_writer_memory(buf, sizeof (buf));
        fprintf(stderr, "length of buffer is --- > %d\n", sizeof (buf));
        avro_write_data(writer, msg_body_schema, body_record);
        avro_writer_flush(writer);
        int64_t data_len = avro_size_data(writer, msg_body_schema, body_record);
        //int64_t data_len  = avro_writer_tell(writer);
        fprintf(stderr, "length of data is --- > %d\n", data_len);

        //for http
        struct curl_slist *plist = NULL;
        plist = curl_slist_append(NULL, "Expect:100-Continue");
        curl_slist_append(plist, "Content-Type: binary/octet-stream");
        curl_slist_append(plist, "Transfer-Encoding:chunked");
        curl_slist_append(plist, "Connection: Keep-Alive");

        CURL * easy_handle = curl_easy_init();
        curl_easy_setopt(easy_handle, CURLOPT_HTTPHEADER, plist);
        curl_easy_setopt(easy_handle, CURLOPT_POSTFIELDS, buf);
        curl_easy_setopt(easy_handle, CURLOPT_POSTFIELDSIZE, data_len);

        curl_easy_setopt(easy_handle, CURLOPT_URL, "http://192.168.11.142:8010/dataload/");
        curl_easy_setopt(easy_handle, CURLOPT_SSL_VERIFYPEER, 0);
        CURLcode res = curl_easy_perform(easy_handle);
        if (res != CURLE_OK) {
            fprintf(stderr, "curl_easy_perform() failed --- > %s\n", curl_easy_strerror(res));
        }

        curl_easy_cleanup(easy_handle);
        curl_slist_free_all(plist); /* free the list again */


        avro_schema_decref(array_schema);
        avro_datum_decref(data_array);

        avro_schema_decref(strusr_map_schema);
        avro_datum_decref(string_map_1);
        avro_schema_decref(string_map_schema);
        avro_datum_decref(string_map);
        avro_datum_decref(map_value);

        avro_datum_decref(body_record);
        avro_schema_decref(msg_body_schema);

        avro_schema_decref(packet_schema);
        avro_schema_decref(msg_body_schema);

        avro_writer_free(writer);
    }
    return 0;
}
/*
 *
 *
 *
 *
 *
 * /
//avro_datum_t value;
/*fprintf(stderr, "avro map get message: %d\n", avro_map_get(string_map_datum, "field1", &value)); */
//char * val;
//avro_string_get(value, &val);
//fprintf(stderr, "avro map get value : %s\n", val);
//
//avro_schema_t string_map_schema= avro_schema_map(avro_schema_string());
//avro_schema_t long_map_schema= avro_schema_map(avro_schema_long());
